/**
 * Copyright 2008-2009. Chongqing Communications Industry Services Co.,Ltd Information Technology Branch. All rights
 * reserved. <a>http://www.cqcis.com</a>
 */
package com.cqcis.uip.base.framework.core.cluster;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.mina.common.IoSession;
import org.osgi.framework.BundleContext;
import org.osgi.framework.ServiceReference;

import com.cqcis.uip.base.framework.FrameworkConstants;
import com.cqcis.uip.base.framework.UIPServiceException;
import com.cqcis.uip.base.framework.UIPTaskException;
import com.cqcis.uip.base.framework.common.HttpParam;
import com.cqcis.uip.base.framework.common.State;
import com.cqcis.uip.base.framework.common.Task;
import com.cqcis.uip.base.framework.common.Task.Source;
import com.cqcis.uip.base.framework.common.Task.Type;
import com.cqcis.uip.base.framework.core.Endpoint;
import com.cqcis.uip.base.framework.core.RegisteredService;
import com.cqcis.uip.base.framework.core.RegisteredServiceFindService;
import com.cqcis.uip.base.framework.core.cluster.ClusterSupportSelector.SwitchInfo;
import com.cqcis.uip.base.framework.utils.EndpointGenerator;

/**
 * 集群 task 处理
 * 
 * @author huangmy
 * @date   2009-1-13
 */
public class ClusterTaskProcessorImpl implements ClusterTaskProcessor {
	private static Log log = LogFactory.getLog(ClusterTaskProcessorImpl.class);
	private BundleContext bundleContext;
	
	public ClusterTaskProcessorImpl(BundleContext bundleContext) {
		this.bundleContext = bundleContext;
	}
	
	/**
	 * @see com.cqcis.uip.base.framework.core.cluster.ClusterTaskProcessor#doTaskSchedule(com.cqcis.uip.base.framework.common.HttpParam, org.apache.mina.common.IoSession)
	 */
	public void doTaskSchedule(HttpParam param, IoSession session) {
		if (log.isDebugEnabled()) {
			log.debug("Param: \n" + param.getHeadersString());
		}
		
		try {
			Type type = ClusterSwitchSupport.getType(param);
			if (log.isDebugEnabled()) {
				log.debug("TYPE: " + type);
			}
			
			if (Type.ACTIVE == type) { // for active
				forActiveProcess(param, session);				
				return;
			}
			
			RegisteredService rs = getRSFor(param);		    
		    if (!rs.scheduleCluster(param)) {
		    	throw new UIPServiceException("scheduleCluster return false, for the RS: " + rs);
		    }
		} catch(Throwable t) {			
			Type type = ClusterSwitchSupport.getType(param);
			if (Type.NORMAL == type || Type.ACTIVE == type) {
				param.setHeaderline("HTTP/1.1 500 Internal Server Error");
				session.write(param);
			}
			log.error("Catch Exception in doTaskSchedule.", t);
		}
	}
	
	/**
	 * 获取RS
	 * 
	 * @param param
	 * @return
	 * @throws Exception
	 */
	private RegisteredService getRSFor(HttpParam param) throws Exception {
		Endpoint endpoint = ClusterSwitchSupport.getToEndpoint(param);
	    if (endpoint == null) {
	    	throw new UIPServiceException("Can't find the endpoint for the Param: " + param);
	    }
	    		    
	    RegisteredService rs = getRegisteredServiceFindService().getRegisteredService(endpoint);
	    if (rs == null) {
	    	throw new UIPServiceException("Can't find the RS for the Endpoint: " + endpoint);
	    }
	    
	    return rs;
	}
	
	/**
	 * FOR ACTIVE
	 * 
	 * @param param
	 * @param session
	 * @throws Exception
	 */
	private void forActiveProcess(HttpParam param, IoSession session) throws Exception {
		Task task = (Task)param.getContents().getObject(Task.class.getClassLoader());				
		if (log.isDebugEnabled()) {
			log.debug("Active Task: \n" + task);
		}
		
		task.setType(Type.RESULT);
		task.setState(State.SUCCESS);
		task.setMessage(Type.ACTIVE.getName());
		
		SwitchInfo si = ClusterSupportSelector.parseFromConstrainKey(task.route().to()
				.getAttribute(Endpoint.LOCATION));
		
		HttpParam resultP = ClusterSwitchSupport.getResponseParams(task, param);
		session.write(resultP).join();// result
		
		// 以session为模板，建立双向client模型
		ClusterSupportSelector css = getClusterSelector();
		css.createOrGetStub(session, si.flip()); // CREATE STUB
		// close 已经复制的原型
		session.close(false).join();
		
		// Task for active
		Task at = new Task(FrameworkConstants.CLUSTER_CT);
		at.setSource(Source.REMOTE);
		at.setType(Type.ACTIVE);
		at.setState(State.SUCCESS);
		Endpoint to = EndpointGenerator.generate(Endpoint.ACTIVE_ID);								
		to.setAttribute(Endpoint.LOCATION, si.getNodeNameA());
		
		at.route().to(to);
		at.route().from(to);
		
		css.schedule(at); // SCHEDULE
	}
	
	/**
	 * selector
	 * 
	 * @return
	 */
	private ClusterSupportSelector getClusterSelector() {
		ClusterSupportSelector css = ClusterSupportSelector.get(this.bundleContext);
		if (css == null) {
			throw new UIPTaskException("Can't find ClusterSupportSelector, may be is not active.");
		}
		
		return css;
	}
	
	/**
	 * 注册服务查找
	 * 
	 * @return
	 */
	private RegisteredServiceFindService getRegisteredServiceFindService() {
		ServiceReference sr = this.bundleContext.getServiceReference(RegisteredServiceFindService.class.getName());
		if (sr != null) {
			return (RegisteredServiceFindService)this.bundleContext.getService(sr);
		}
		
		return null;
	}
}
